package com.shengzai.flink.sink

import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo3MySink {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentsDS: DataStream[String] = env.readTextFile("data/students.txt")

    studentsDS.addSink(new Sinks)

    studentsDS.print()

    env.execute()
  }


}


class Sinks extends SinkFunction[String]{
  /**
   *invoke： 每一条数据执行一次
   * @param value 每一行数据
   * @param context 上下文对象
   */
  override def invoke(value: String, context: SinkFunction.Context): Unit = {
    println(s"自定义sink:$value")
  }
}

